import threading
import json
import asyncio
from asyncio import AbstractEventLoop
from time import time, sleep
import sys
import os
from decimal import Decimal
import traceback
import logging

from django.core.management.base import BaseCommand
from django.conf import settings
from django.db import transaction, close_old_connections

from channels.layers import get_channel_layer

from django_redis import get_redis_connection
from bitmex.models import Account

from bitmex.api.bitmex_ws import BitMEXWebsocket
log = logging.getLogger(__name__)

logging.getLogger("aioredis").setLevel(logging.WARNING)

if settings.ASYNCIO_DEBUG:
    AbstractEventLoop.set_debug()


def redis_cli():
    return get_redis_connection('default')

class AccountClient(threading.Thread):

    def __init__(self, account, use_proxy):
        super(AccountClient, self).__init__()
        self.conns = {}
        self.account = Account.objects.get(name=account)
        self.bm = BitMEXWebsocket(self.account.api_key, self.account.api_secret,
                test=self.account.is_testnet,
                sub_account_msg=True, use_proxy=use_proxy)
        self.bm.connect()
        sleep(3)
        self.bm.set_on_msg_callbacks(self.process_message)

    @staticmethod
    def start_loop(loop):
        asyncio.set_event_loop(loop)
        loop.run_forever()

    def process_message(self, message):
        """处理实时交易信息。把信息推送给订阅的客户端
        """
        asyncio.run_coroutine_threadsafe(self._process_message(message), self.loop)

    async def _process_message(self, message):
        """处理市场上实时交易数据，直接推送至客户端
        """
        # 获得在线客户端列表,获得1分钟内活跃的客户端
        cli = redis_cli()
        key = f'subscribers:{self.account.id}:bitmex'
        print(key)
        channels = cli.zremrangebyscore(key, 0, time() - 60)  # 移除60秒内不活跃的客户端
        channels = cli.zrange(key, 0, -1)  # 获取剩下所有客户端
        # 遍历客户端，推送消息
        print(channels)
        print(message)
        channel_layer = get_channel_layer()
        for channel in channels:
            channel = channel.decode('utf-8')
            await channel_layer.send(channel, {
                'type': 'update.message',
                'message': message
            })

    def run(self):
        """
        通过django channel 获取需要订阅的bitmex acccount topic，然后让bitmex client
        订阅它
        :return:
        """
        self.loop = asyncio.new_event_loop()
        t = threading.Thread(target=self.start_loop, args=(self.loop,))
        t.setDaemon(True)
        t.start()
        # 主要用来接受中断，同时可以让其起的线程也跟着中断
        while True:
            cli = redis_cli()
            result = cli.blpop(f'bitmex:user:{self.account.id}', timeout=10)
            if not result:
                continue
            item = result[1].decode('utf-8')
            try:
                item = json.loads(item)
                topic = item['topic']
                log.info(time() - item['time'])
                if time() - item['time'] > 60:
                    log.info('old message, discard it')
                    continue  # 1分钟以前的信息丢弃
                if item['action'] == 'subscribe':
                    log.info(f'subscribe {topic}')
                    if topic in self.conns:
                        continue
                    conn = self.conns.get(topic, None)
                    if not conn:
                        self.bm.sub_topic(topic)
                        self.conns[topic] = True
                elif item['action'] == 'unsubscribe':
                    log.info('unsubscribe!!')
            except Exception:
                log.info('error occur')
                tp, val, tb = sys.exc_info()
                log.info(tp)
                log.info(val)
                traceback.print_tb(tb)


class SymbolClient(threading.Thread):

    def __init__(self, use_proxy):
        super(SymbolClient, self).__init__()
        self.conns = {}
        self.bm = BitMEXWebsocket("IvayHLyekSP3Ptw7j2JcSj1s", "z9Zxw2_Ly7nX0UvFkpaezNI-KMgh5SE1itTRLtKzyV6CkYKq",
                test=False,
                sub_account_msg=False, use_proxy=use_proxy)
        self.bm.connect()
        sleep(3)
        self.bm.set_on_msg_callbacks(self.process_message)

    @property
    def stream(self):
        return 'bitmex_symbol'

    def parse_message(self, message):
        if "trade" in message or "orderBook" in message:
            return message

    @staticmethod
    def start_loop(loop):
        asyncio.set_event_loop(loop)
        loop.run_forever()

    def process_message(self, message):
        """处理实时交易信息。把信息推送给订阅的客户端
        """
        asyncio.run_coroutine_threadsafe(self._process_message(message), self.loop)

    async def _process_message(self, message):
        """处理市场上实时交易数据，直接推送至客户端
        """

        # 获得在线客户端列表,获得1分钟内活跃的客户端
        cli = redis_cli()
        # key = f'subscribers:{self.stream}:{symbol.upper()}:bitmex'
        key = f'subscribers:{self.stream}:bitmex'
        channels = cli.zremrangebyscore(key, 0, time() - 60)  # 移除60秒内不活跃的客户端
        channels = cli.zrange(key, 0, -1)  # 获取剩下所有客户端

        # 转换消息格式
        msg_obj = self.parse_message(message)
        # 遍历客户端，推送消息
        print(channels)
        channel_layer = get_channel_layer()
        if msg_obj:
            for channel in channels:
                channel = channel.decode('utf-8')
                await channel_layer.send(channel, {
                    'type': 'update.message',
                    'message': msg_obj
                })

    def run(self):
        """获取市场上实时交易。
        维护正在监听的市场进程，有需要时停止之。
        从事件队列中获取新的监听需求。
        """
        self.loop = asyncio.new_event_loop()
        t = threading.Thread(target=self.start_loop, args=(self.loop,))
        t.setDaemon(True)
        t.start()
        while True:
            cli = redis_cli()
            result = cli.blpop(f'event_bus:bitmex:{self.stream}', timeout=10)
            if not result:
                continue
            item = result[1].decode('utf-8')
            try:
                item = json.loads(item)
                symbol = item['symbol']
                log.info(time() - item['time'])
                if time() - item['time'] > 60:
                    log.info('old message, discard it')
                    continue  # 1分钟以前的信息丢弃
                if item['action'] == 'subscribe':
                    log.info('subscribe !!!')
                    if symbol in self.conns:
                        continue
                    conn = self.conns.get(symbol, None)
                    if not conn:
                        self.bm.sub_symbol(symbol)
                        self.conns[symbol] = True
                elif item['action'] == 'unsubscribe':
                    log.info('unsubscribe!!')
            except Exception:
                log.info('error occur')
                tp, val, tb = sys.exc_info()
                log.info(tp)
                log.info(val)
                traceback.print_tb(tb)


class Command(BaseCommand):
    """运行币安的websocket客户端。
    工作：
    1. 一个固定的币安帐号去获取市场价格、报价等。
    2. 每个币安帐号只监听自己的用户信息的变更即可。
    """
    help = '运行币安websocket客户端'

    def add_arguments(self, parser):
        parser.add_argument(
            '--action', type=str, default=['symbol', 'account'],
            choices=['symbol', 'account'], nargs='+',
            help='指定运行的websocket客户端类型, 可选 symbol, account'
        )
        parser.add_argument(
            '--proxy', type=bool, default=True, help="是否使用代理"
        )

        parser.add_argument(
            '--account', type=str, required=False, help='币安帐号'
        )

    def handle(self, *args, **options):
        """根据不同的参数，有三种运行方式：
        1. 获取实时币价。
        2. 获取市场报价深度数据。
        3. 交易帐号实时更新。
        """
        action = options['action']
        use_proxy = options["proxy"]
        if 'symbol' in action:
            log.info('starting up symbol client')
            SymbolClient(use_proxy).start()
        if 'account' in action:
            log.info('starting up account client')
            account = options["account"]
            AccountClient(account, use_proxy).start()
